package org.artifact.core.plugin.rpc.client;

import cn.hutool.cache.impl.TimedCache;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.artifact.core.plugin.rpc.codec.RpcDecoder;
import org.artifact.core.plugin.rpc.codec.RpcEncoder;
import org.artifact.core.plugin.rpc.pojo.RpcRequest;
import org.artifact.core.plugin.rpc.pojo.RpcResponse;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * RPC 客户端（用于发送 RPC 请求）
 *
 * @author huangyong
 * @since 1.0.0
 */
public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Log log = LogFactory.get(RpcClient.class);
    private String host;
    private int port;
    private Bootstrap bootstrap;
    private EventLoopGroup group;
    private Channel channel;
    // AsyncCompletionStage
    private TimedCache<String, CompletableFuture<RpcResponse>> synchRespMap = null;

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;

        synchRespMap = new TimedCache<>(3000);
        synchRespMap.schedulePrune(1000);

        group = new NioEventLoopGroup();
        // 创建并初始化 Netty 客户端 Bootstrap 对象
        bootstrap = new Bootstrap();
        bootstrap.group(group);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new RpcEncoder(RpcRequest.class)); // 编码 RPC 请求
                pipeline.addLast(new RpcDecoder(RpcResponse.class)); // 解码 RPC 响应
                pipeline.addLast(RpcClient.this); // 处理 RPC 响应
            }
        });
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        // 连接 RPC 服务器
        connect();
    }

    private void connect(){
        try {
            channel = bootstrap.connect(host, port).sync().channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        CompletableFuture<RpcResponse> completableFuture = synchRespMap.get(response.getRequestId());
        if (completableFuture!=null){
            completableFuture.complete(response);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("api caught exception", cause);
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        String key = inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort();
        RpcProxy.clients.remove(key);
    }

    public RpcResponse send(RpcRequest request) throws Throwable {
        CompletableFuture<RpcResponse> completableFuture = synchRespMap.get(request.getRequestId(),() -> new CompletableFuture<RpcResponse>());
        channel.writeAndFlush(request);
        RpcResponse response = completableFuture.get(3, TimeUnit.SECONDS);
        return response;
    }
    
    public void stop() {
    	try {
			group.shutdownGracefully().sync();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
    }
}
